package com.gitee.xhs;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author BigHead
 * @date 2022/6/29 10:17
 * @description FlinkEnvironmentFactory
 */
public class FlinkEnvironmentFactory {

    /**
     * 本地执行生成Web页面
     *
     * @param environmentConfiguration 配置类
     * @return StreamExecutionEnvironment
     */
    public static StreamExecutionEnvironment webUi(EnvironmentConfig environmentConfiguration) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(environmentConfiguration.getConfiguration());
        checkpointEnvironment(env, environmentConfiguration);
        return env;
    }

    /**
     * 本地执行生成Web页面
     *
     * @param environmentConfiguration 配置类
     * @return StreamExecutionEnvironment
     */
    public static StreamExecutionEnvironment simpleEnv(EnvironmentConfig environmentConfiguration) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(environmentConfiguration.getConfiguration());
        checkpointEnvironment(env, environmentConfiguration);
        return env;
    }

    private static void checkpointEnvironment(StreamExecutionEnvironment env, EnvironmentConfig environmentConfiguration) {
        env.setParallelism(environmentConfiguration.getParallelism());
        if (environmentConfiguration.getUseCheckPoint()) {
            env.enableCheckpointing(environmentConfiguration.getCheckpointInterval());

            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointStorage(environmentConfiguration.getCheckPointFilePath());
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(environmentConfiguration.getExternalizedCheckpointCleanup());
        }
    }

}
